package com.lyon.custom.server.service;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.NamedThreadFactory;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.lyon.custom.api.model.ConfigListenerReqBO;
import com.lyon.custom.api.model.ConfigPublishReqVO;
import com.lyon.custom.server.common.notify.NotifyCenter;
import com.lyon.custom.server.common.notify.data.LocalDataChangeEvent;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletResponse;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author LeeYan9
 * @since 2021-11-29
 */
@Service
@AllArgsConstructor
public class LongPollingService {

    @Autowired
    private PersistenceService persistenceService;

    public LongPollingService() {
        this.subscribers = new ConcurrentLinkedQueue<>();
        // 添加配置变更监听器， 提前返回配置变更groupKeys;
        NotifyCenter.addSubscriber(event -> {
            if (LocalDataChangeEvent.class.isAssignableFrom(event.getClass())) {
                Executors
                        .newSingleThreadExecutor(new NamedThreadFactory("DataChangeTask-", true))
                        .execute(new DataChangeTask((LocalDataChangeEvent) event));
            }
        }, LocalDataChangeEvent.class);
    }

    public void publishConfig(ConfigPublishReqVO configPublishReqVO) {
        //保存数据
        persistenceService.insertOrUpdate(configPublishReqVO);

        // 发布配置改变事件 #同步集群的配置;
//        NotifyCenter.publishEvent(new ConfigDataChangeEvent(configPublishReqVO));
    }

    public void doPollingClient(ConfigListenerReqBO configListenerReqBO) {
        AsyncContext asyncContext = configListenerReqBO.getRequest().startAsync();
        asyncContext.setTimeout(0);
        Executors.newSingleThreadExecutor().execute(new LongPollingClient(asyncContext, configListenerReqBO.getPollingTimeout(), configListenerReqBO.getClientMd5Map()));
    }


    class DataChangeTask implements Runnable {

        ConfigPublishReqVO publishReqVO;

        public DataChangeTask(LocalDataChangeEvent event) {
            this.publishReqVO = event.getSource();
        }

        @Override
        public void run() {
            // 获取 id;
            String tenantId = publishReqVO.getTenantId();
            String group = publishReqVO.getGroup();
            String dataId = publishReqVO.getDataId();
            String groupKey = StrUtil.join(";", tenantId, group, dataId);
            Iterator<LongPollingClient> iterator = subscribers.iterator();
            while (iterator.hasNext()) {
                LongPollingClient subscriber = iterator.next();
                // 订阅了当前配置
                if (subscriber.clientMd5Map.containsKey(groupKey)) {
                    // 返回变更的groupKey
                    subscriber.sendResponse(List.of(groupKey));
                    // 删除当前订阅
                    iterator.remove();
                }
            }
        }
    }

    Queue<LongPollingClient> subscribers;

    @Data
    class LongPollingClient implements Runnable {
        private AsyncContext asyncContext;
        private Integer pollingTimeout;
        private Map<String, String> clientMd5Map;
        private ScheduledFuture<?> scheduledFuture;

        public LongPollingClient(AsyncContext asyncContext, Integer pollingTimeout, Map<String, String> clientMd5Map) {
            this.asyncContext = asyncContext;
            this.pollingTimeout = pollingTimeout;
            this.clientMd5Map = clientMd5Map;
        }

        // 等待超时时间之前执行 , 获取变更配置
        @Override
        public void run() {
            ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("LongPollingClient-", true));
            this.scheduledFuture = executorService.schedule(() -> {
                // 手动执行，比较Md5;获取变更配置groupKey;
                List<String> changeKeys = new ArrayList<>();
                try {
                    if (clientMd5Map != null) {
                        Iterator<Map.Entry<String, String>> iterator = clientMd5Map.entrySet().iterator();
                        while (iterator.hasNext()) {
                            Map.Entry<String, String> entry = iterator.next();
                            String groupKey = entry.getKey();
                            String md5 = entry.getValue();
                            ConfigCacheService.ConfigItem configItem = ConfigCacheService.getItem(groupKey, null, null);
                            if (configItem != null && !StrUtil.equals(configItem.getMd5(), md5)) {
                                changeKeys.add(groupKey);
                            }
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                subscribers.remove(LongPollingClient.this);
                // 通知 客户端;
                sendResponse(changeKeys);
            }, pollingTimeout - 500, TimeUnit.MILLISECONDS);
            // 方便本地配置变更时, 关闭当前调度;
            subscribers.add(this);
        }

        void sendResponse(List<String> changeKeys) {
            // 取消调度.
            if (null != scheduledFuture) {
                scheduledFuture.cancel(false);
            }
            if (CollUtil.isEmpty(changeKeys)) {
                asyncContext.complete();
                return;
            }
            try {
                HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();
                // Disable cache.
                response.setHeader("Pragma", "no-cache");
                response.setDateHeader("Expires", 0);
                response.setHeader("Cache-Control", "no-cache,no-store");
                response.setStatus(HttpServletResponse.SC_OK);
                response.getWriter().println(JSONUtil.toJsonStr(changeKeys));
            } catch (Throwable e) {
                e.printStackTrace();
            } finally {
                asyncContext.complete();
            }
        }
    }

}
